package com.shujia;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.fs.permission.FsPermission;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;

/*
        java操作HDFS来创建，上传，下载，删除，重命名，查看文件的描述信息。

        进阶一下：使用IO流的方式进行上传和下载
 */
public class JavaToHdfs {
    public static void main(String[] args) throws IOException {
        //1、获取HDFS的文件系统
        Configuration conf = new Configuration();
        //指定连接集群的主节点，以及端口号
        conf.set("fs.defaultFS", "hdfs://master:9000");
        //如果想要操作HDFS,就必须获取到HDFS提供的抽象父类的实现，FileSystem
        FileSystem fs = FileSystem.get(conf);
        System.out.println(fs);

        //创建文件夹
//        boolean b = fs.mkdirs(new Path("/shujia/bigdata17"));
        //如果文件夹已经存在，则创建失败，返回false
//        System.out.println(b);

        //文件的上传
        //第一个Path指的是本地的文件路径
        //第二个Path指的是HDFS上的路径
//        fs.copyFromLocalFile(
//                new Path("E:\\projects\\IdeaProjects\\hadoop-bigdata17\\hadoop-hdfs\\data\\words.txt"),
//                new Path("/shujia/bigdata17/"));

        //文件的下载
//        fs.copyToLocalFile(new Path("/shujia/bigdata17/students.txt"),
//                new Path("E:\\shujia\\十七期\\hadoop\\day01\\students2.txt"));

        //文件的删除
//        boolean b = fs.delete(new Path("/shujia/bigdata17/students.txt"),true);
//        System.out.println(b);

        //文件重命名
//        boolean b = fs.rename(new Path("/shujia/bigdata17/students.txt"), new Path("/shujia/bigdata17/students2.txt"));
//        System.out.println(b);

        //查看文件的描述信息。
        RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/shujia/bigdata17"), true);
        System.out.println("/shujia/bigdata17目录下文件的相关信息");
        while (listFiles.hasNext()) {
            LocatedFileStatus status = listFiles.next();
            //获取文件的名字
            String name = status.getPath().getName();
            System.out.println("文件的名称：" + name);
            //获取文件的长度
            long len = status.getLen();
            System.out.println("文件的字节数：" + len);
            //查看文件所属用户
            FsPermission permission = status.getPermission();
            System.out.println("文件的权限：" + permission);
            String owner = status.getOwner();
            System.out.println("文件的所属用户：" + owner);
            String group = status.getGroup();
            System.out.println("分组信息：" + group);
            long time = status.getModificationTime();
            System.out.println(time);

            //查看存储块的信息
            BlockLocation[] locations = status.getBlockLocations();
            for (BlockLocation blockLocation : locations) {
                //获取块对应存储的位置信息，主机节点
                String[] hosts = blockLocation.getHosts();
                for (String s : hosts) {
                    System.out.println(s);
                }
            }

            System.out.println("===");
        }


        //使用IO流操作进行文件的上传
        //创建输入流，读取本地的文件
        //不能写相对路径
        FileInputStream fis = new FileInputStream(new File("E:\\projects\\IdeaProjects\\hadoop-bigdata17\\hadoop-hdfs\\data\\students.txt"));
        //创建输出流 FSDataOutputStream
        FSDataOutputStream fsDataOutputStream = fs.create(new Path("/test/students.txt"));
        //使用流进行拷贝,流对拷贝
        IOUtils.copy(fis,fsDataOutputStream);
        //释放资源
        IOUtils.closeQuietly(fsDataOutputStream);
        IOUtils.closeQuietly(fis);


        //使用IO流操作进行文件的下载
        FSDataInputStream fsDataInputStream = fs.open(new Path("/test/students.txt"));
        //创建输出流
        FileOutputStream fos = new FileOutputStream(new File("E:\\shujia\\十七期\\hadoop\\day02\\students.txt"));
        //使用流进行拷贝,流对拷贝
        IOUtils.copy(fsDataInputStream,fos);
        //释放资源
        IOUtils.closeQuietly(fsDataInputStream);
        IOUtils.closeQuietly(fos);


        //释放资源
        fs.close();

    }
}
